package com.hsc.www.rawTest.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class RPCServer {

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setVirtualHost("my_vhost");
        connectionFactory.setPort(5672);
        // 创建一个连接
        Connection connection = connectionFactory.newConnection();
        // 创建一个频道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(RPC_QUEUE_NAME, true, false, false, null);
        //限制：每次最多给一个消费者发送1条消息
        channel.basicQos(1);
        //为rpc_queue队列创建消费者，用于处理请求
        channel.basicConsume(RPC_QUEUE_NAME, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                        .correlationId(properties.getCorrelationId())
                        .build();
                //获取回调队列名字
                String callQueueName = properties.getReplyTo();
                String receiveMessage = new String(body, "UTF-8");

                System.out.println("receiveMessage =" + receiveMessage + "    ReplyTo = " + properties.getReplyTo() + "    CorrelationId = " + properties.getCorrelationId());
                String[] args = receiveMessage.split(",");
                String message=(Integer.valueOf(args[0])+Integer.valueOf(args[1]))+"";

                //先发送回调结果
                channel.basicPublish("", callQueueName, replyProps, message.getBytes("UTF-8"));
                //后手动发送消息反馈
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        System.out.println("SERVER START......");

        while (true){
            TimeUnit.SECONDS.sleep(1);
        }
    }
}